package com.fafi.rxjava2.observeon.backpress;

import com.fafi.rxjava1.subscribeon.Switcher;
import com.fafi.rxjava2.backpress.Drop;
import com.fafi.rxjava2.backpress.Receiver;
import com.fafi.rxjava2.operatorbackpress.Telephoner;
import com.fafi.rxjava2.operatorbackpress.TelephonerWithUpstream;

import java.util.LinkedList;
import java.util.Queue;

public class TelephonerCallbackOn<T> extends TelephonerWithUpstream<T, T> {

    private final Switcher mSwitcher;

    public TelephonerCallbackOn(Telephoner<T> source, Switcher mSwitcher) {
        super(source);
        this.mSwitcher = mSwitcher;
    }

    @Override
    protected void callActual(Receiver<T> receiver) {
        source.call(new CallbackOnReceiver<>(receiver, mSwitcher));
    }

    private static final class CallbackOnReceiver<T> implements Receiver<T>, Runnable {
        private final Receiver<T> tReceiver;
        private final Switcher.Worker worker;
        private final Queue<T> tQueue = new LinkedList<>();

        public CallbackOnReceiver(Receiver<T> tReceiver, Switcher switcher) {
            this.tReceiver = tReceiver;
            this.worker = switcher.createWorker();
        }

        @Override
        public void onCall(Drop d) {
            tReceiver.onCall(d);
        }

        @Override
        public void onReceive(T t) {
            tQueue.offer(t);
            switches();
        }


        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onCompleted() {

        }

        @Override
        public void run() {
            T t = tQueue.poll();
            tReceiver.onReceive(t);

        }

        private void switches() {
            worker.switches(this);
        }
    }
}
